package transform;

import beans.SenSorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.Collections;

public class CollectAndColMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //从文件中读取数据
        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SenSorReading> mapStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SenSorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        //分流，将数据按照三十度来进行分流
        SplitStream<SenSorReading> splitStream = mapStream.split(new OutputSelector<SenSorReading>() {
            @Override
            public Iterable<String> select(SenSorReading senSorReading) {
                return (senSorReading.getTemperature() > 30 ? Collections.singletonList("high") : Collections.singletonList("low"));
            }
        });
        DataStream<SenSorReading> highStream = splitStream.select("high");
        DataStream<SenSorReading> lowStream = splitStream.select("low");
        DataStream<SenSorReading> allStream = splitStream.select("high", "low");

//        highStream.print("high");
//        lowStream.print("low");
//        allStream.print("all");

        /**
         * 合流
         * 将高温转换成二元组类型，于低温流连接之后，输出状态信息
         */
        SingleOutputStreamOperator<Tuple2<String, Double>> highMapStream = highStream.map(new MapFunction<SenSorReading, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(SenSorReading senSorReading) throws Exception {
                return new Tuple2<>(senSorReading.getId(), senSorReading.getTemperature());
            }
        });

        //连接
        ConnectedStreams<Tuple2<String, Double>, SenSorReading> connectStream = highMapStream.connect(lowStream);
        /**
         * 三个参数的意义
         * 前两个分别是合并map的类型
         * 第三个是需要转换成的数据类型
         */
        SingleOutputStreamOperator<Object> map = connectStream.map(new CoMapFunction<Tuple2<String, Double>, SenSorReading, Object>() {
            @Override
            public Object map1(Tuple2<String, Double> stringDoubleTuple2) throws Exception {
                return new Tuple3<>(stringDoubleTuple2.f0,stringDoubleTuple2.f1,"high temp waring");
            }

            @Override
            public Object map2(SenSorReading senSorReading) throws Exception {
                return new Tuple2<>(senSorReading.getId(),"normal");
            }
        });

      map.print();

        /**
         * union联合多条liu,但是数据类型必须一致
         */
        DataStream<SenSorReading> union = highStream.union(lowStream, allStream);

        env.execute();
    }
}
